﻿using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace WebAPI.Utils
{
    /// <summary>
    /// RabbitMQ.Client
    /// 消息积压指的是当消息的生产速度远超过消费速度时，大量消息在队列中等待被处理的现象。这种情况可能导致系统延迟增加、资源占用过高甚至服务不可用等问题。
    /// </summary>
    internal class RabbitMqHelper
    {
        private static IConnection? _connection;
        
        private static IModel? Channel { get; set; }
        
        private const string SimpleQueue = "simple_queue";

        private const string DelayExchange = "delay_exchange";
        
        private const string DelayQueue = "delay_queue";
        
        public static void Configure()
        {
            // 创建连接工厂
            var factory = new ConnectionFactory
            {
                HostName = "localhost",
                //UserName = "sa",
                //Password = "sa@123",
                //VirtualHost = "vhost"
            };

            // 创建连接
            _connection = factory.CreateConnection();

            // 创建通道
            Channel = _connection.CreateModel();
            
            // 简单模式
            Channel.QueueDeclare(queue: DelayQueue, durable: true, exclusive: false, autoDelete: false, arguments: null);

            // 延迟队列
            var arguments = new Dictionary<string, object>
            {
                {"x-delayed-type", "direct"}//direct topic
            };
            // 声明普通/延迟交换机 Tcpic / Fanout(发布订阅) / Direct / x-delayed-message
            Channel.ExchangeDeclare(exchange: DelayExchange, type: "x-delayed-message", durable: true, autoDelete: false, arguments: arguments);// arguments 普通：null
            Channel.QueueDeclare(queue: DelayQueue, durable: true, exclusive: false, autoDelete: false, arguments: null);
            Channel.QueueBind(queue: DelayQueue, exchange: DelayExchange, routingKey: "", arguments: null);
        }
        
        // 模式
        // 简单模式
        public static void SimpleServer()
        {
            var consumer = new EventingBasicConsumer(Channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine($" [x] Received {message}");
            };
            // 开始消费队列中的消息
            Channel.BasicConsume(queue: SimpleQueue,
                autoAck: true,
                consumer: consumer);
        }
        
        public static void SimpleClient(string message)
        {
            var body = Encoding.UTF8.GetBytes(message);
            Channel.BasicPublish(exchange: string.Empty, routingKey: SimpleQueue, basicProperties: null, body: body);
        }
        
        // 工作队列模式
        public void WorkClient(string message)
        {
            Channel.QueueDeclare(queue: "task_queue",
                     durable: true,
                     exclusive: false,
                     autoDelete: false,
                     arguments: null);

            var body = Encoding.UTF8.GetBytes(message);

            var properties = Channel.CreateBasicProperties();
            properties.Persistent = true;

            Channel.BasicPublish(exchange: string.Empty,
                                 routingKey: "task_queue",
                                 basicProperties: properties,
                                 body: body);
        }

        public void WorkServer()
        {
            Channel.QueueDeclare(queue: "task_queue",
                     durable: true,
                     exclusive: false,
                     autoDelete: false,
                     arguments: null);

            Channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

            Console.WriteLine(" [*] Waiting for messages.");

            var consumer = new EventingBasicConsumer(Channel);
            
            consumer.Received += (model, ea) =>
            {
                byte[] body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine($" [x] Received {message}");

                int dots = message.Split('.').Length - 1;

                Console.WriteLine(" [x] Done");

                // here channel could also be accessed as ((EventingBasicConsumer)sender).Model
                Channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            };
            Channel.BasicConsume(queue: "task_queue",
                                 autoAck: false,
                                 consumer: consumer);
        }

        // 发布订阅模式
        public void SubPubClient(string message)
        {
            Channel.ExchangeDeclare(exchange: "logs", type: ExchangeType.Fanout);

            var body = Encoding.UTF8.GetBytes(message);
            Channel.BasicPublish(exchange: "logs",
                                 routingKey: string.Empty,
                                 basicProperties: null,
                                 body: body);
            Console.WriteLine($" [x] Sent {message}");

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

        public void SubPubServer()
        {
            Channel.ExchangeDeclare(exchange: "logs", type: ExchangeType.Fanout);

            // declare a server-named queue
            var queueName = Channel.QueueDeclare().QueueName;
            Channel.QueueBind(queue: queueName,
                              exchange: "logs",
                              routingKey: string.Empty);

            Console.WriteLine(" [*] Waiting for logs.");

            var consumer = new EventingBasicConsumer(Channel);
            consumer.Received += (model, ea) =>
            {
                byte[] body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine($" [x] {message}");
            };
            Channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

        // 路由模式
        public void RouteClient(string severity, string message)
        {
            Channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);

            var body = Encoding.UTF8.GetBytes(message);
            Channel.BasicPublish(exchange: "direct_logs",
                                 routingKey: severity,
                                 basicProperties: null,
                                 body: body);
            Console.WriteLine($" [x] Sent '{severity}':'{message}'");

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

        public void RouteServer()
        {
            Channel.ExchangeDeclare(exchange: "direct_logs", type: ExchangeType.Direct);
            // declare a server-named queue
            var queueName = Channel.QueueDeclare().QueueName;

            var severities = new List<string>();
            foreach (var severity in severities)
            {
                Channel.QueueBind(queue: queueName,
                                  exchange: "direct_logs",
                                  routingKey: severity);
            }

            Console.WriteLine(" [*] Waiting for messages.");

            var consumer = new EventingBasicConsumer(Channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                var routingKey = ea.RoutingKey;
                Console.WriteLine($" [x] Received '{routingKey}':'{message}'");
            };
            Channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

        // 通配符模式
        public void TopicClient(string routingKey, string message)
        {
            Channel.ExchangeDeclare(exchange: "topic_logs", type: ExchangeType.Topic);

            var body = Encoding.UTF8.GetBytes(message);
            Channel.BasicPublish(exchange: "topic_logs",
                                 routingKey: routingKey,
                                 basicProperties: null,
                                 body: body);
            Console.WriteLine($" [x] Sent '{routingKey}':'{message}'");
        }

        public void TopicServer()
        {
            Channel.ExchangeDeclare(exchange: "topic_logs", type: ExchangeType.Topic);
            // declare a server-named queue
            var queueName = Channel.QueueDeclare().QueueName;


            var severities = new List<string>();
            foreach (var bindingKey in severities)
            {
                Channel.QueueBind(queue: queueName,
                                  exchange: "topic_logs",
                                  routingKey: bindingKey);
            }

            Console.WriteLine(" [*] Waiting for messages. To exit press CTRL+C");

            var consumer = new EventingBasicConsumer(Channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                var routingKey = ea.RoutingKey;
                Console.WriteLine($" [x] Received '{routingKey}':'{message}'");
            };
            Channel.BasicConsume(queue: queueName,
                                 autoAck: true,
                                 consumer: consumer);

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }

        // RPC
        
        
        // 延迟队列
        public static void DelayServer()
        {
            // 事件基本消费者
            var consumer = new EventingBasicConsumer(Channel);
            // 接收到消息事件
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine("Received message: {0}", message);
                LoggerHelper.Info(message);
                // 手动应答确认消息被消费
                Channel.BasicAck(ea.DeliveryTag, false);
            };
            // 设置手动应答消息 
            Channel.BasicConsume(queue: DelayQueue, autoAck: false, consumer: consumer);
        }

        public static void DelayClient(string message, int delaySeconds = 0)
        {
            var delayTime = TimeSpan.FromSeconds(delaySeconds);
            var props = Channel.CreateBasicProperties();
            props.Headers = new Dictionary<string, object> { { "x-delay", delayTime.TotalMilliseconds } };
            var body = Encoding.UTF8.GetBytes(message);
            // Fanout类型无需指定routingKey
            Channel.BasicPublish(exchange: DelayExchange, routingKey: "", basicProperties: props, body: body);
        }
    }
 }
